package demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.Random;

@Slf4j
public class RealTimeProducer {
    public static void main(String[] args) {
        // 设置Kafka连接属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.40.10.187:6667"); // Kafka服务器地址
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建KafkaProducer实例
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            // 创建随机数生成器
            Random random = new Random();

            // 定义订单类别
            List<String> orderTypes = Arrays.asList("图书", "游戏", "服装", "电子", "家电");
            String title = "Generated and sent data:订单类型\t产品名称\t单价\t订单日期\t是否上架";
            log.info("标题:{}", title);

            // 数据生成和推送
            while (true) {
                // 从预定义的订单类别中随机选择一个
                String orderType = orderTypes.get(random.nextInt(orderTypes.size()));

                // 生成随机数据
                String orderName = "Product" + random.nextInt(5);
                int orderQuantity = random.nextInt(10) + 1;
                String orderDate = "2024" + String.format("%02d", random.nextInt(12) + 1) + String.format("%02d", random.nextInt(30) + 1);
                String isValid = random.nextBoolean() ? "Y" : "N";

                // 构建数据字符串
                String data = orderType + "\t" + orderName + "\t" + orderQuantity + "\t" + orderDate + "\t" + isValid;

                // 创建ProducerRecord并发送数据
                ProducerRecord<String, String> record = new ProducerRecord<>("orders", null, data);

                // 发送消息并设置回调函数
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            // 如果有异常，打印错误信息
                            System.err.println("Failed to send message due to: " + exception.getMessage());
                        }
                        // 移除了成功的回调输出
                    }
                });

                // 输出数据到控制台
                System.out.println("Generated and sent data: " + data);

                // 每隔5秒推送一次数据
                try {
                    Thread.sleep(5 * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    break; // 如果线程被中断，退出循环
                }
            }
        } // 使用try-with-resources自动关闭producer
    }
}